package com.fwmagic.flink.projectcase.redissink.sink;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

public class MyRedisSink extends RichSinkFunction<Tuple3<String,String,String>> {

    private transient Jedis jedis;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        ParameterTool parameterTool = (ParameterTool)getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        String host = parameterTool.get("redis.host");
        String passowrd = parameterTool.get("redis.passowrd");
        int port = parameterTool.getInt("redis.port",6379);
        int db = parameterTool.getInt("redis.db");
        jedis = new Jedis(host, port, 2000);
        if(StringUtils.isNotBlank(passowrd)){
            jedis.auth(passowrd);
        }
        jedis.select(db);
    }

    @Override
    public void invoke(Tuple3<String, String, String> value, Context context) {
        jedis.hset(value.f0, value.f1, value.f2);
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }
}
